package com.ylkj.bus.dispatcher.handler;

import com.hwl.datahub.DataHubWriteConfig;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.device.report.UpdateDevicMonitorAndUnpack;
import com.ylkj.bus.dispatcher.IStreamHandler;
import com.ylkj.bus.monitor.equipment.model.EquipmentAlarmMonitorLogEnty;
import com.ylkj.bus.onoffline.report.ReportOnlineStatusHandler;
import com.ylkj.io.stream.rds.RdsConfig;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/4/23
 * @Desc:
 */
@RequiredArgsConstructor
@Data
public class OnlineAlarmDataStreamHandler implements IStreamHandler {
    private final RdsConfig rdsConfig;
    private final String rdsTableName;
    private final String rdsTableName2;
    private final RedisConfig redisConfig;
    private final String rdsTableName3;
    private final String rdsTableName4;
    @Override
    public void handle(DataStream<MessageContent> stream) {
        // 定义侧输出流标签
        OutputTag<EquipmentAlarmMonitorLogEnty> outTagStatus = new OutputTag<EquipmentAlarmMonitorLogEnty>("outTagStatus") {};
        //定义主流
        SingleOutputStreamOperator<MessageContent> mainProcessDs = stream.process(new ReportOnlineStatusHandler(redisConfig, outTagStatus));
        //获取侧输出流
        DataStream<EquipmentAlarmMonitorLogEnty> sideOutput = mainProcessDs.getSideOutput(outTagStatus);
        sideOutput.print();
        //更新设备监控表
        sideOutput.process(new UpdateDevicMonitorAndUnpack(rdsTableName, rdsConfig,redisConfig,rdsTableName2,rdsTableName3,rdsTableName4)).name(rdsTableName + "_update_process");

    }
}
